#include <asm/types.h>
#include <linux/posix_types.h>
#include <sys/types.h>

typedef __u64 uint64_t;

#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <math.h>
#include <pthread.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#include "sfr.h"

#include "boxmuller.h"

//-----------------------------------------------------------------------------

//#define NEW_STAT_FPRINTF

#define error_fprintf(x ...) fprintf(x)
#ifndef NEW_STAT_FPRINTF
#define  stat_fprintf(x ...) do { if  (stat_mode) fprintf(x); } while (0)
#endif
#define debug_fprintf(x ...) do { if (debug_mode) fprintf(x); } while (0)
#define trace_fprintf(x ...) do { if (trace_mode) fprintf(x); } while (0)

char           *program_name            = "generator";

int             quiet_mode              = 0;
int             stat_mode               = 0;
int             debug_mode              = 0;
int             trace_mode              = 0;

char           *error_heading           = "Error";
char           *warning_heading         = "Warning";
char           *debug_heading           = "Debug";
char           *trace_heading           = "Trace";

int             worker_count_max        = 16;
int             worker_count            = 0;
int             exit_before_doing_work  = 0;

pthread_mutex_t atomic_mutex;

struct timespec clock_start;
struct timespec clock_stop;

int             argv_check(void);
int             argv_parse_uint64_with_suffix(char **argv, uint64_t *value);
int             argv_parse(int argc, char *argv[]);
void            argv_print(void);
void            usage_show(void);

int             atomic_fetch_and_add_int(int *value, int addend);
uint64_t        atomic_fetch_and_add_uint64(uint64_t *value, uint64_t addend);
void            atomic_fetch_int_and_uint64(int *var0, int *out0, uint64_t *var1, uint64_t *out1);

int             mutex_init(pthread_mutex_t *mutex);
void            mutex_lock(pthread_mutex_t *mutex);
void            mutex_unlock(pthread_mutex_t *mutex);

char          **path_list_parse(char *path_list, char separator, int *array_size);

int             work_init(void);
int             work_perform(void);

int             thread_create(void *context, void *(thread_entry)(void *));

void            worker_start(void);
void            worker_stop(void);
int             worker_count_get(void);

#ifdef NEW_STAT_FPRINTF
void            stat_fprintf(FILE *fp, const char *fmt, ...);
#endif

//-----------------------------------------------------------------------------

#define BUFFER_MAX (1024 * 1024)

#define FILE_TYPE_NORMAL 1
#define FILE_TYPE_XFR    2

#define LOG_MODE_UNIFIED    1
#define LOG_MODE_PER_THREAD 2

#define LOG_STAT_BUFFER_MAX (1024 * 1024 * 4)
#define LOG_STATUS_LINE_BUFFER_MAX (1024)

typedef struct {
    char *file;
    char *path;
    char *http;
    int   size;
    int   type;
    int   hash;
} task_t;

typedef struct {
    pthread_t  pthread;
    int        id;
    int        file_done_count;
} worker_t;

typedef struct {
    int  size;
    int  used;
    int  head;
    int  tail;
    int  data[0];
} queue_t;

typedef struct {
    pthread_mutex_t  ready_mutex;
    pthread_mutex_t  empty_mutex;
    int              value;
} mailbox_t;

char           *log_path_format         = "/var/tmp/log.generator.%d";

char           *base_path_list          = "/mnt";
char          **base_path_array         = NULL;
int             base_path_array_size    = 0;

char           *http_path_list          = "http://myserver.com/testfiles";
char          **http_path_array         = NULL;
int             http_path_array_size    = 0;

char            path_list_separator     = ',';

int             file_type               = FILE_TYPE_NORMAL;

double          file_size_dev           = 8192.0;
double          file_size_avg           = 32768.0;

int             file_count_max          = 0;

uint64_t        nfr_file_size_limit     = 1024ULL * 1024ULL * 1024ULL;
uint64_t        nfr_file_size_saturate  = 0;

uint64_t        xfr_file_size_limit     = 1024ULL * 1024ULL;
uint64_t        xfr_file_size_saturate  = 1;

uint64_t        total_size_max          = 1024ULL * 1024ULL * 1024ULL;
uint64_t        total_size              = 0;
uint64_t        total_size_done         = 0;

int             per_directory_count_max = 4096;

int             file_count              = 0;
int             file_count_done         = 0;

float           stat_interval           = 30.0;

int             log_mode                = LOG_MODE_UNIFIED;

int             unified_log_fd          = -1;

int             swap_file_dir1          = 1;

int             boxmuller_seed          = 1;
pthread_mutex_t boxmuller_mutex;

mailbox_t      *mailbox = NULL;

int             filler_started          = 0;

unsigned long   random_seed             = 1;

int             consecutive_zero_rate_max = 20;

char            file_buffer[BUFFER_MAX];

int             log_buffer_used = 0;

char            log_buffer[LOG_STAT_BUFFER_MAX];

int             file_id_next(void);

void           *filler_thread_entry(void *context);
void           *worker_thread_entry(void *context);
void            worker(int worker_id, int log_fd);

task_t         *task_next(int worker_id);
void            task_free(task_t *task);
int             task_next_file_size(int file_id);
char           *task_next_file_name(int file_id);
char           *task_next_path_name(task_t *task);
char           *task_next_http_name(task_t *task);

int             path_parent_create(char *path);

void            mailbox_init(int worker_count);
void            mailbox_put(int worker_id, int value);
int             mailbox_get(int worker_id);

//
// Function:    main()
// Purpose:     Program entry.
// Return:      0 on successful parse.
//              -1 on any failure.
//
int main(int argc, char **argv)
{
    assert(argc > 0);
    assert(argv);
    assert(*argv);

    program_name = *argv;

    if (argv_parse(argc, argv) < 0) {
        usage_show();
        return -1;
    }

    if (argv_check() < 0) {
        return -1;
    }

    if (!quiet_mode)
        argv_print();

    if (exit_before_doing_work)
        return 0;

    if (work_init() < 0) {
        error_fprintf(stderr, "%s: work_init() aborted\n", error_heading);
        return -1;
    }

    if (work_perform() < 0) {
        error_fprintf(stderr, "%s: work_perform() aborted\n", error_heading);
        return -1;
    }

    return 0;
}

//
// Function:    argv_parse_uint64_with_suffix()
// Purpose:     Parse an integer with an optional suffix.
// Return:      0 on successful parse.
//              -1 on any failure.
//
int argv_parse_uint64_with_suffix(char **argv, uint64_t *value)
{
    uint64_t  size;
    char     *suffix;

    assert(argv);
    assert(*argv);
    assert(value);

    size = strtoll(*argv, &suffix, 0);
    switch (*suffix) {
        case 't':
        case 'T':
            size = size * 1024LL;
        case 'g':
        case 'G':
            size = size * 1024LL;
        case 'm':
        case 'M':
            size = size * 1024LL;
        case 'k':
        case 'K':
            size = size * 1024LL;
            debug_fprintf(stderr, "%s: argv_parse_uint64_with_suffix() -> %lld\n", debug_heading, (long long int)size);
        case '\0':
            *value = size;
            return 0;
        default:
            return -1;
    }
}

//
// Function:    argv_parse()
// Purpose:     Parse the command line arguments.
// Return:      0 on successful parse.
//              -1 on any failure.
//
int argv_parse(int argc, char **argv)
{
    char     *total_suffix;
    uint64_t  value;

    assert(argc > 0);
    assert(argv);

    if (argc == 1) {
        return -1;
    }

    while (*++argv) {
        if (argv[0][0] != '-') {
            return -1;
        }

        switch (argv[0][1]) {

        case '\0':
            return -1;

        case 'c':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &total_size_max) < 0) {
                return -1;
            }
            break;

        case 'f':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &value) < 0) {
                return -1;
            }
            file_count_max = value;
            break;

        case 'F':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &value) < 0) {
                return -1;
            }
            file_count = value;
            break;

        case 'w':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &value) < 0) {
                return -1;
            }
            per_directory_count_max = value;
            break;

        case 'A':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &value) < 0) {
                return -1;
            }
            file_size_avg = value;
            break;

        case 'S':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &value) < 0) {
                return -1;
            }
            file_size_dev = value;
            break;

        case 't':
            ++argv;
            if (!*argv) {
                return -1;
            }
            if (argv_parse_uint64_with_suffix(argv, &value) < 0) {
                return -1;
            }
            worker_count_max = value;
            break;

        case 'n':
            file_type = FILE_TYPE_NORMAL;
            break;

        case 'x':
            file_type = FILE_TYPE_XFR;
            break;

        case 'b':
            ++argv;
            if (!*argv) {
                return -1;
            }
            base_path_list = strdup(*argv);
            break;

        case 'h':
            ++argv;
            if (!*argv) {
                return -1;
            }
            http_path_list = strdup(*argv);
            break;

        case 'L':
            ++argv;
            if (!*argv) {
                return -1;
            }
            log_path_format = *argv;
            break;

        case 'p':
            log_mode = LOG_MODE_PER_THREAD;
            break;

        case 'u':
            log_mode = LOG_MODE_UNIFIED;
            break;

        case 'q':
            quiet_mode = 1;
            break;

        case 'v':
            if (debug_mode) {
                trace_mode = 1;
            } else {
                debug_mode = 1;
            }
            break;

        case 'V':
            stat_mode = 1;
            break;

        case 'X':
            exit_before_doing_work = 1;
            break;

        default:
            return -1;
        }
    }

    base_path_array = path_list_parse(base_path_list, path_list_separator, &base_path_array_size);
    if (base_path_array == NULL) {
        return -1;
    }

    http_path_array = path_list_parse(http_path_list, path_list_separator, &http_path_array_size);
    if (http_path_array == NULL) {
        return -1;
    }

    return 0;
}

//
// Function:    argv_check()
// Purpose:     Make sure all of the work parameters are reasonable.
//
int argv_check(void)
{
    if (base_path_array_size == 0) {
        error_fprintf(stderr, "%s: base_path_list is empty\n", error_heading);
        return -1;
    }

    if (http_path_array_size == 0) {
        error_fprintf(stderr, "%s: http_path_list is empty\n", error_heading);
        return -1;
    }

    if (http_path_array_size != base_path_array_size) {
        error_fprintf(stderr, "%s: base_path_list does not have the same number of paths as http_path_list\n", error_heading);
        return -1;
    }

    return 0;
}

//
// Function:    argv_print()
// Purpose:     Display all of the work parameters.
//
void argv_print(void)
{
    int index;

    printf("log_path_format         = \"%s\"\n", log_path_format);

    printf("base_path_array         = \"%s\"\n", base_path_array[0]);
    for (index = 1; index < base_path_array_size; index++) {
        printf("                        = \"%s\"\n", base_path_array[index]);
    }

    printf("http_path_array         = \"%s\"\n", http_path_array[0]);
    for (index = 1; index < http_path_array_size; index++) {
        printf("                        = \"%s\"\n", http_path_array[index]);
    }

    printf("file_type               = %s\n", (file_type == FILE_TYPE_NORMAL) ? "NORMAL (POSIX)" : "XFR");
    printf("file_size_avg           = %.1f\n", file_size_avg);
    printf("file_size_dev           = %.1f\n", file_size_dev);
    printf("file_count (start)      = %d\n", file_count);
    printf("file_count_max          = %d%s\n", file_count_max, (file_count_max == 0) ? " (no limit)" : "");
    printf("total_size              = %lld\n", (long long int)total_size);
    printf("total_size_max          = %lld\n", (long long int)total_size_max);
    printf("per_directory_count_max = %d\n", per_directory_count_max);
    printf("worker_count_max        = %d\n", worker_count_max);
    printf("log_mode                = %s\n", (log_mode == LOG_MODE_UNIFIED) ? "UNIFIED" : "PER-THREAD");
    printf("stat_mode               = %s\n", stat_mode ? "ON" : "OFF");
    printf("stat_interval           = %f\n", stat_interval);
}

//
// Function:    usage_show()
// Purpose:     Display the proper usage of the command.
//
void usage_show(void)
{
    error_fprintf(stderr, "Usage: %s [ options ]\n", program_name);
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -c bytes      Approximate maximum number of bytes to write (default: 1G)\n");
    error_fprintf(stderr, "          -f files      Maximum number of files to create (default: 0 which means no limit)\n");
    error_fprintf(stderr, "          -w width      Directory fanout (default: 4K)\n");
    error_fprintf(stderr, "          -A bytes      File size average (default: 32K)\n");
    error_fprintf(stderr, "          -S bytes      File size standard deviation (default: 8K)\n");
    error_fprintf(stderr, "          -P char       Path list separator character (default: ,)\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -t count      Number of worker threads (default: 16)\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -n            Write to normal (POSIX) file system (default)\n");
    error_fprintf(stderr, "          -x            Write to XFR\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -b pathlist   Where to write the data files (default: /mnt)\n");
    error_fprintf(stderr, "          -h pathlist   Artificial http path prefix written to log file (default: http://myserver.com/testfiles)\n");
    error_fprintf(stderr, "          -L path       Specify the path (printf format) where logs are written (default: /var/tmp/log.generator.%%d)\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -p            Per-thread logging\n");
    error_fprintf(stderr, "          -u            Unified logging (default)\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "Debug:\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -q            Do not display session parameters\n");
    error_fprintf(stderr, "          -v            Verbose debug mode\n");
    error_fprintf(stderr, "          -V            Verbose statistics mode\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "          -F count      File count start (default: 0)\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "Notes:\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "1. Numeric values can have a suffix of \"K\", \"M\" or \"G\".\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "2. \"path\" for -L can have a \"%%d\" specifying the thread number.\n");
    error_fprintf(stderr, "   (This is useful when using per-thread logging).\n");
    error_fprintf(stderr, "\n");
    error_fprintf(stderr, "3. \"pathlist\" is a list of comma separated paths.  The separator\n");
    error_fprintf(stderr, "   can be changed from comma to another character using -P.  Please\n");
    error_fprintf(stderr, "   Note that the http pathlist and the base pathlist are expected to\n");
    error_fprintf(stderr, "   correlate, the 1st entry in each list are matched up, etc ...\n");
}

//
// Function:    atomic_fetch_and_add_int()
// Purpose:     Atomically fetch and add to a variable.
// Return:      The old value of the variable.
//
int atomic_fetch_and_add_int(int *value, int addend)
{
    int old_value;

    assert(value);

    mutex_lock(&atomic_mutex);
    old_value = *value;
    *value += addend;
    mutex_unlock(&atomic_mutex);

    return old_value;
}

//
// Function:    atomic_fetch_and_add_uint64()
// Purpose:     Atomically fetch and add to a variable.
// Return:      The old value of the variable.
//
uint64_t atomic_fetch_and_add_uint64(uint64_t *value, uint64_t addend)
{
    uint64_t old_value;

    assert(value);

    mutex_lock(&atomic_mutex);
    old_value = *value;
    *value += addend;
    mutex_unlock(&atomic_mutex);

    return old_value;
}

//
// Function:    atomic_fetch_int_and_uint64()
// Purpose:     Atomically fetch capture the values of two variables.
//
void atomic_fetch_int_and_uint64(int *var0, int *out0, uint64_t *var1, uint64_t *out1)
{
    assert(var0);
    assert(out0);
    assert(var1);
    assert(out1);

    mutex_lock(&atomic_mutex);
    *out0 = *var0;
    *out1 = *var1;
    mutex_unlock(&atomic_mutex);
}

//
// Function:    mutex_init()
// Purpose:     Initialize a mutex.
// Return:      0 on success, -1 on failure.
//
int mutex_init(pthread_mutex_t *mutex)
{
    pthread_mutexattr_t mutex_attributes;
	
    assert(mutex);

    if (pthread_mutexattr_init(&mutex_attributes) != 0) {
        return -1;
    }

    if (pthread_mutex_init(mutex, &mutex_attributes) != 0) {
        return -1;
    }

    return 0;
}

//
// Function:    mutex_lock()
// Purpose:     Lock a mutex.
//
void mutex_lock(pthread_mutex_t *mutex)
{
    int rc;

    assert(mutex);

    rc = pthread_mutex_lock(mutex);
    assert(rc == 0);
}

//
// Function:    mutex_unlock()
// Purpose:     Unlock a mutex.
//
void mutex_unlock(pthread_mutex_t *mutex)
{
    int rc;

    assert(mutex);

    rc = pthread_mutex_unlock(mutex);
    assert(rc == 0);
}

//
// Function:    path_list_parse()
//
char **path_list_parse(char *path_list, char separator, int *array_size)
{
    int    index = 0;
    char **array;
    char  *ptr;

    assert(path_list);
    assert(array_size);

    *array_size = 1;
    for (ptr = path_list; *ptr; ptr++) {
        if (*ptr == separator) {
            (*array_size)++;
        }
    }

    array = calloc(*array_size + 1 + 1, sizeof(char *));
    if (!array) {
        error_fprintf(stderr, "%s: path_list_parse(): Out of memory\n", error_heading);
        return NULL;
    }

    array[index++] = &path_list[0];
    for (ptr = path_list; *ptr; ptr++) {
        if (*ptr == separator) {
            array[index++] = &ptr[1];
            *ptr = '\0';
        }
    }

    return array;
}

//
// Function:    work_init()
// Purpose:     Primary entry point to start working.
//
int work_init(void)
{
    int rc;

    trace_fprintf(stderr, "%s: work_init()\n", trace_heading);

    rc = mutex_init(&atomic_mutex);
    assert(rc == 0);

    rc = mutex_init(&boxmuller_mutex);
    assert(rc == 0);

    boxmuller_init(boxmuller_seed);

    mailbox_init(worker_count_max);

    if (log_mode == LOG_MODE_UNIFIED) {
        char log_path[PATH_MAX];

        sprintf(log_path, log_path_format, 0);
        unified_log_fd = open(log_path, O_CREAT | O_TRUNC | O_RDWR | O_APPEND, 0644);
        if (unified_log_fd < 0) {
            error_fprintf(stderr, "%s: open(\"%s\"): failed (%s)\n", error_heading, log_path, strerror(errno));
            assert(0);
            return -1;
        }
    }

    return 0;
}

//
// Function:    work_perform()
// Purpose:     Primary entry point to start working.
//
int work_perform(void)
{
    int thread_index;
    int new_worker_count;
    int old_worker_count;
    int workers_remaining;
    int curr_file_count_done;
    int prev_file_count_done = 0;
    int diff_file_count_done = 0;
    float curr_file_rate;
    float total_file_rate;
    int consecutive_zero_rate = 0;
    uint64_t curr_total_size_done;
    uint64_t prev_total_size_done = 0;
    uint64_t diff_total_size_done;
    float curr_byte_rate;
    float total_byte_rate;
    worker_t *worker;
    struct timespec clock_curr;
    struct timespec clock_prev;
    float diff_sec;
    float total_sec;

    trace_fprintf(stderr, "%s: work_perform()\n", trace_heading);

    memset(file_buffer, 0, BUFFER_MAX);

    clock_gettime(CLOCK_REALTIME, &clock_start);

    for (thread_index = 0; thread_index < worker_count_max; thread_index++) {
        worker = calloc(1, sizeof(*worker));
        if (!worker) {
            error_fprintf(stderr, "%s: calloc(): worker_t failed\n", error_heading);
            return -1;
        }
        worker->id = thread_index;
        old_worker_count = worker_count_get();
        if (thread_create(worker, worker_thread_entry) < 0) {
            error_fprintf(stderr, "%s: thread_create(worker): failed\n", error_heading);
            return -1;
        }
        do {
            usleep(1);
            new_worker_count = worker_count_get();
        } while (new_worker_count == old_worker_count);
    }

    if (thread_create(worker, filler_thread_entry) < 0) {
        error_fprintf(stderr, "%s: thread_create(filler): failed\n", error_heading);
        return -1;
    }
    do {
        usleep(1);
    } while (!filler_started);

    clock_prev = clock_start;

    while (1) {
        workers_remaining = worker_count_get();

        trace_fprintf(stderr, "%s: workers_remaining = %d\n", trace_heading, workers_remaining);

        sleep(1);

        //
        // Capture the current values.
        //
        atomic_fetch_int_and_uint64(&file_count_done, &curr_file_count_done, &total_size_done, &curr_total_size_done);
        clock_gettime(CLOCK_REALTIME, &clock_curr);

        //
        // Get elapsed time intervals
        //
        diff_sec = ((float)(clock_curr.tv_sec - clock_prev.tv_sec)) + 0.000000001 * ((float)(clock_curr.tv_nsec - clock_prev.tv_nsec));
        total_sec = ((float)(clock_curr.tv_sec - clock_start.tv_sec)) + 0.000000001 * ((float)(clock_curr.tv_nsec - clock_start.tv_nsec));

        if ((diff_sec >= stat_interval) || (workers_remaining == 0)) {
            diff_file_count_done = curr_file_count_done - prev_file_count_done;
            diff_total_size_done = curr_total_size_done - prev_total_size_done;

            curr_file_rate = ((float) diff_file_count_done) / diff_sec;
            curr_byte_rate = ((float) diff_total_size_done) / diff_sec;

            total_file_rate = ((float) curr_file_count_done) / total_sec;
            total_byte_rate = ((float) curr_total_size_done) / total_sec;

            stat_fprintf(stderr, "threads=%d curr={ %f sec, %d files, %lld bytes, %f files/sec, %f bytes/sec } total={ %f sec, %d files, %lld bytes, %f files/sec, %f bytes/sec }\n",
                         workers_remaining,
                         diff_sec, diff_file_count_done, (long long int)diff_total_size_done, curr_file_rate, curr_byte_rate,
                         total_sec, curr_file_count_done, (long long int)curr_total_size_done, total_file_rate, total_byte_rate);

            if (!(curr_file_rate > 0.0)) {
                consecutive_zero_rate++;
                if (consecutive_zero_rate >= consecutive_zero_rate_max) {
                    log_buffer[log_buffer_used] = '\0';
                    fprintf(stderr, "%s\n", log_buffer);
                    assert(0);
                }
            } else {
                consecutive_zero_rate = 0;
            }

            prev_file_count_done = file_count_done;
            prev_total_size_done = total_size_done;

            clock_prev = clock_curr;
        }

        if (workers_remaining <= 0)
            break;
    }

    clock_gettime(CLOCK_REALTIME, &clock_stop);

    sleep(1);

    log_buffer[log_buffer_used] = '\0';
    fprintf(stderr, "%s\n", log_buffer);

    return 0;
}

//
// Function:    thread_create()
// Purpose:     Create a single worker thread.
//
int thread_create(void *context, void *(thread_entry)(void *))
{
    pthread_attr_t attributes;
    pthread_t      thread_id;

    trace_fprintf(stderr, "%s: thread_create()\n", trace_heading);

    if (pthread_attr_init(&attributes) != 0) {
        error_fprintf(stderr, "%s: pthread_attr_init() failed\n", error_heading);
        return -1;
    }

    if (pthread_attr_setdetachstate(&attributes, PTHREAD_CREATE_DETACHED) != 0) {
        error_fprintf(stderr, "%s: pthread_attr_setdetachstate() failed\n", error_heading);
        return -1;
    }

    if (pthread_attr_setscope(&attributes, PTHREAD_SCOPE_SYSTEM) != 0) {
        error_fprintf(stderr, "%s: pthread_attr_setscope() failed\n", error_heading);
        return -1;
    }

    if (pthread_create(&thread_id, &attributes, thread_entry, (void *)context) != 0) {
        error_fprintf(stderr, "%s: pthread_create() failed\n", error_heading);
        return -1;
    }

    return 0;
}

//
// Function:    worker_start()
// Purpose:     ...
//
void worker_start(void)
{
    atomic_fetch_and_add_int(&worker_count, 1);
}

//
// Function:    worker_stop()
// Purpose:     ...
//
void worker_stop(void)
{
    atomic_fetch_and_add_int(&worker_count, -1);
}

//
// Function:    worker_count()
// Purpose:     ...
//
int worker_count_get(void)
{
    return worker_count;
}

#ifdef NEW_STAT_FPRINTF
//
// Function:    stat_fprintf()
// Purpose:     ...
//
void stat_fprintf(FILE *fp, const char *fmt, ...)
{
    va_list  ap;
    int      len;
    int      off;
    char     str[LOG_STATUS_LINE_BUFFER_MAX];

    assert(fp);
    assert(fmt);

    if (!stat_mode)
        return;

    va_start(ap, fmt);
    vsnprintf(str, LOG_STATUS_LINE_BUFFER_MAX, fmt, ap);
    va_end(ap);

    len = strlen(str);

    off = atomic_fetch_and_add_int(&log_buffer_used, len);

    if ((off + len) < LOG_STATUS_LINE_BUFFER_MAX) {
        memcpy(&log_buffer[off], str, len);
    } else {
        atomic_fetch_and_add_int(&log_buffer_used, -len);
    }
}
#endif // NEW_STAT_FPRINTF

//
// Function:    filler_thread_entry()
// Purpose:     Entry point of the filler thread.
//
void *filler_thread_entry(void *context)
{
    int worker_id;

    filler_started = 1;

    while (1) {
        for (worker_id = 0; worker_id < worker_count_max; worker_id++) {
            mailbox_put(worker_id, file_id_next());
        }
    }
}

//
// Function:    worker_thread_entry()
// Purpose:     Entry point of single worker thread.
//
void *worker_thread_entry(void *context)
{
    int       worker_id;
    worker_t *w;
    int       log_fd;
    char      log_path[PATH_MAX];

    worker_start();

    w = context;

    worker_id = w->id;

    debug_fprintf(stderr, "%s: worker %d: start\n", debug_heading, worker_id);

    if (log_mode == LOG_MODE_PER_THREAD) {
        sprintf(log_path, log_path_format, worker_id);
        log_fd = open(log_path, O_CREAT | O_TRUNC | O_RDWR | O_APPEND, 0644);
        if (log_fd < 0) {
            error_fprintf(stderr, "%s: worker %d: open(\"%s\"): failed (%s)\n", error_heading, worker_id, log_path, strerror(errno));
            assert(0);
            goto exit;
        }
    } else {
        assert(unified_log_fd >= 0);
        log_fd = unified_log_fd;
    }

    worker(worker_id, log_fd);

exit:

    if (log_mode == LOG_MODE_PER_THREAD) {
        if (log_fd >= 0)
            close(log_fd);
    }

    debug_fprintf(stderr, "%s: worker %d: stop\n", debug_heading, worker_id);

    sleep(1);

    worker_stop();

    pthread_exit(NULL);

    return NULL; // Should never get here.
}

//
// Function:    worker()
// Purpose:     Perform the core of the worker tasks.
//
void worker(int worker_id, int log_fd)
{
    task_t *task;
    int     file_rate;
    int     file_fd;
    int     file_size;
    int     write_size;
    int     files_done;
    struct timespec clock_curr;
    char   *file_suffix = ".txt";
//  char    file_buffer[BUFFER_MAX];
    char    http_buffer[PATH_MAX + 1];

//  memset(file_buffer, 0, BUFFER_MAX);

    while ((task = task_next(worker_id))) {
        file_size = task->size;

        switch (task->type) {

        case FILE_TYPE_NORMAL:
            if (path_parent_create(task->path) < 0) {
                error_fprintf(stderr, "%s: worker %d: mkdir(\"%s\"): failed (%s)\n", error_heading, worker_id, task->path, strerror(errno));
                task_free(task);
                return;
            }

            trace_fprintf(stderr, "%s: worker %d: open(\"%s\")\n", trace_heading, worker_id, task->path);

            file_fd = open(task->path, O_CREAT | O_TRUNC | O_RDWR, 0666);
            if (file_fd < 0) {
                error_fprintf(stderr, "%s: worker %d: open(\"%s\"): failed (%s)\n", error_heading, worker_id, task->path, strerror(errno));
                assert(0);
                task_free(task);
                return;
            }

            while (file_size > 0) {
                write_size = (file_size > BUFFER_MAX) ? BUFFER_MAX : file_size;
                if (write(file_fd, file_buffer, write_size) < 0) {
                    error_fprintf(stderr, "%s: worker %d: write(\"%s\"): failed (%s)\n", error_heading, worker_id, task->path, strerror(errno));
                    close(file_fd);
                    task_free(task);
                    return;
                }
                file_size -= write_size;
            }

            close(file_fd);

            sprintf(http_buffer, "%s\n", task->http);

            break;

#ifdef __linux__
        case FILE_TYPE_XFR:
            task->file = SFR_Write(base_path_array[0], file_buffer, write_size, file_suffix);
            if (!(task->file)) {
                error_fprintf(stderr, "%s: worker %d: SFR_Write(): failed (%s)\n", error_heading, worker_id, strerror(errno));
                task_free(task);
                return;
            }

            task->path = task_next_path_name(task);
            if (!task->path) {
                free(task);
                return;
            }

            task->http = task_next_http_name(task);
            if (!task->http) {
                free(task);
                return;
            }

            sprintf(http_buffer, "%s\n", task->http);

            break;
#endif

        default:
            assert(0);
        }

        atomic_fetch_and_add_uint64(&total_size_done, task->size);

        atomic_fetch_and_add_int(&file_count_done, 1);

        clock_gettime(CLOCK_REALTIME, &clock_curr);

        debug_fprintf(stderr, "%s: worker %d: %lld bytes: %s type: %s (%d files/sec)\n",
                      debug_heading,
                      worker_id,
                      (long long int)(task->size),
                      task->http,
                      (task->type == FILE_TYPE_NORMAL) ? "nfr" : (task->type == FILE_TYPE_XFR) ? "xfr" : "???",
                      file_rate);

        if (write(log_fd, http_buffer, strlen(http_buffer)) < 0) {
            error_fprintf(stderr, "%s: worker %d: write(LOG_PATH): failed (%s)\n", error_heading, worker_id, strerror(errno));
            task_free(task);
            return;
        }

        task_free(task);
    }

    return;
}

//
// Function:    file_id_next()
// Purpose:     Fetch the next file ID value used to generate the next file name.
// Return:      File id.
//
int file_id_next(void)
{
    int file_id;

    file_id = atomic_fetch_and_add_int(&file_count, 1);

    return file_id;
}

//
// Function:    task_next()
// Purpose:     Fetch the next task to perform.
// Return:      Structure describing the next task.
//
task_t *task_next(int worker_id)
{
    int       file_id;
    uint64_t  old_total_size;
    task_t   *task;

    task = calloc(1, sizeof(task_t));
    if (!task) {
        return NULL;
    }

    debug_fprintf(stderr, "%s: file_count=%d\n",
                  debug_heading,
                  file_count);

    file_id = mailbox_get(worker_id);

    if (file_count_max) {
        if (file_id >= file_count_max) {
            task_free(task);
            return NULL;
        }
    }

    task->hash = random();

    task->file = task_next_file_name(file_id);
    if (!task->file) {
        task_free(task);
        return NULL;
    }

    task->path = task_next_path_name(task);
    if (!task->path) {
        task_free(task);
        return NULL;
    }

    task->http = task_next_http_name(task);
    if (!task->http) {
        task_free(task);
        return NULL;
    }

    task->size = task_next_file_size(file_id);

    task->type = file_type;

    if (nfr_file_size_saturate) {
        if (task->type == FILE_TYPE_NORMAL) {
            if (task->size >= nfr_file_size_limit) {
                task->size = nfr_file_size_limit;
            }
        }
    }

    if (xfr_file_size_saturate) {
        if (task->type == FILE_TYPE_XFR) {
            if (task->size >= xfr_file_size_limit) {
                task->size = xfr_file_size_limit;
            }
        }
    }

    old_total_size = atomic_fetch_and_add_uint64(&total_size, task->size);
    if (total_size_max && (old_total_size >= total_size_max)) {
        atomic_fetch_and_add_uint64(&total_size, -(task->size));
        task_free(task);
        return NULL;
    }

    return task;
}

//
// Function:    task_free()
// Purpose:     Free up the task structure.
//
void task_free(task_t *task)
{
    assert(task);

    if (task->file)
        free(task->file);
    task->file = NULL;

    if (task->path)
        free(task->path);
    task->path = NULL;

    if (task->http)
        free(task->http);
    task->http = NULL;

    free(task);
}

//
// Function:    task_next_file_size()
// Purpose:     Generate the file size for the next file.
//
int task_next_file_size(int file_id)
{
    int file_size;

    mutex_lock(&boxmuller_mutex);

    file_size = boxmuller(file_size_avg, file_size_dev);

    mutex_unlock(&boxmuller_mutex);

    if (file_size < 0) {
        file_size = -file_size;
    }

    return file_size;
}

//
// Function:    task_next_file_name()
// Purpose:     Generate the file name for the next file.
//
char *task_next_file_name(int file_id)
{
    int   dir1;
    int   dir2;
    int   dir3;
    int   file;
    char *path;

    path = malloc(PATH_MAX + 1);
    if (!path) {
        return NULL;
    }

    file = file_id % per_directory_count_max;
    file_id = file_id / per_directory_count_max;
    dir1 = file_id % per_directory_count_max;
    file_id = file_id / per_directory_count_max;
    dir2 = file_id % per_directory_count_max;
    file_id = file_id / per_directory_count_max;
    dir3 = file_id % per_directory_count_max;

    if (swap_file_dir1) {
        sprintf(path,
                "dir%04d/dir%04d/dir%04d/file%04d",
                dir3,
                dir2,
                file,
                dir1);
    } else {
        sprintf(path,
                "dir%04d/dir%04d/dir%04d/file%04d",
                dir3,
                dir2,
                dir1,
                file);
    }

    debug_fprintf(stderr, "%s: file_id=%d path=\"%s\"\n",
                  debug_heading,
                  file_id,
                  path);

    return path;
}

//
// Function:    task_next_path_name()
// Purpose:     Generate the path name for the next file.
//
char *task_next_path_name(task_t *task)
{
    char *path;

    assert(task);
    assert(task->file);

    path = malloc(PATH_MAX + 1);
    if (!path) {
        return NULL;
    }

    sprintf(path, "%s/%s", base_path_array[task->hash % base_path_array_size], task->file);

    return path;
}

//
// Function:    task_next_http_name()
// Purpose:     Generate the http path name for the next file.
//
char *task_next_http_name(task_t *task)
{
    char *path;

    assert(task);
    assert(task->file);

    path = malloc(PATH_MAX + 1);
    if (!path) {
        return NULL;
    }

    sprintf(path, "%s%s%s", http_path_array[task->hash % http_path_array_size], (task->type == FILE_TYPE_XFR) ? "" : "/", task->file);

    return path;
}

//
// Function:    path_parent_create()
// Purpose:     Create the parent directory path.
// Return:      0 on success, -1 on failure.
// Warning:     This function is a stack and memory hog!
//
int path_parent_create(char *path)
{
    char *pointer;
    char  buffer[PATH_MAX];

    assert(path);

    debug_fprintf(stderr, "%s: path_parent_create(\"%s\")\n", debug_heading, path);

    strcpy(buffer, path);
    pointer = strrchr(buffer, '/');
    if (!pointer) {
        return 0;
    }

    *pointer = '\0';
    if (mkdir(buffer, 0777) == 0) {
        return 0;
    }

    if (errno == EEXIST) {
        return 0;
    }

    if (errno != ENOENT) {
        return -1;
    }

    if (path_parent_create(buffer) < 0) {
        return -1;
    }

    if (mkdir(buffer, 0777) == 0) {
        return 0;
    }

    if (errno == EEXIST) {
        return 0;
    }

    return -1;
}

//
// Function:    mailbox_init()
// Purpose:     Initialize the mailbox sub-system and pre-fill all mailboxes.
//
void mailbox_init(int worker_count)
{
    int worker_id;
    int queue_depth;

    assert(worker_count > 0);

    mailbox = calloc(worker_count, sizeof(mailbox_t));
    assert(mailbox);

    for (worker_id = 0; worker_id < worker_count_max; worker_id++) {
        mutex_init(&(mailbox[worker_id].ready_mutex));
        mutex_init(&(mailbox[worker_id].empty_mutex));
        mutex_lock(&(mailbox[worker_id].ready_mutex));
    }
}

//
// Function:     mailbox_put()
// Purpose:      Fetch a non-zero value from the mailbox for this worker.  If
//               there isn't anything in the mailbox, then the value returned
//               is zero.
//
void mailbox_put(int worker_id, int value)
{
    assert(worker_id < worker_count_max);
    assert(mailbox);

    mutex_lock(&mailbox[worker_id].empty_mutex);

    mailbox[worker_id].value = value;

    debug_fprintf(stderr, "%s: mailbox_put(%d, %d)\n", debug_heading, worker_id, value);

    mutex_unlock(&mailbox[worker_id].ready_mutex);
}

//
// Function:     mailbox_get()
// Purpose:      Fetch a non-zero value from the mailbox for this worker.  If
//               there isn't anything in the mailbox, then the value returned
//               is zero.
//
int mailbox_get(int worker_id)
{
    int value;

    assert(worker_id < worker_count_max);
    assert(mailbox);

    mutex_lock(&mailbox[worker_id].ready_mutex);

    value = mailbox[worker_id].value;

    debug_fprintf(stderr, "%s: mailbox_get(%d) -> %d\n", debug_heading, worker_id, value);

    mutex_unlock(&mailbox[worker_id].empty_mutex);

    return value;
}
